package cn.doitedu.flink.rules;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Timer;

public class Demo {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> sds = env.socketTextStream("localhost", 9989);

        SingleOutputStreamOperator<String> oper = sds.map(new RichMapFunction<String, String>() {
            RuleBean ruleBean = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                ruleBean = new RuleBean();
                Timer timer = new Timer();
                timer.schedule(new LoadRuleRunnable(ruleBean), 0, 2000);
            }

            @Override
            public String map(String value) throws Exception {
                return value + " -> " + ruleBean.getFlag();
            }
        });

        oper.print();
        env.execute("job");
    }
}
